//package com.bsj.travel.device.consumer.rocketmq;
//
//import cn.hutool.http.HttpUtil;
//import com.alibaba.fastjson2.JSON;
//import com.bsj.travel.cached.redis.RedisCached;
//import com.bsj.travel.config.rocketmq.constant.ConsumerGroupConstant;
//import com.bsj.travel.constant.BaseServerTopicConstants;
//import com.bsj.travel.constant.MqttTopicPowerConstants;
//import com.bsj.travel.constant.NumericalConstants;
//import com.bsj.travel.constants.RedisConstant;
//import com.bsj.travel.def.common.DO.DeviceDO;
//import com.bsj.travel.def.common.DO.SubscribeDO;
//import com.bsj.travel.def.common.VO.device.DeviceAttributeVO;
//import com.bsj.travel.def.common.VO.device.HttpMessagePushVO;
//import com.bsj.travel.def.global.mq.Rp;
//import com.bsj.travel.util.ExceptionUtil;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.lang3.StringUtils;
//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.apache.rocketmq.spring.annotation.MessageModel;
//import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
//import org.apache.rocketmq.spring.core.RocketMQListener;
//import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.scheduling.annotation.Async;
//import org.springframework.stereotype.Component;
//
//import java.util.Arrays;
//import java.util.List;
//import java.util.Map;
//import java.util.stream.Collectors;
//
///**
// * @author LiJiaXin
// * @version 1.0.0
// * @description ProductSubscribeMessageRocketmqConsumer
// * @newTime 2024/2/29 16:51
// */
//@Slf4j
//@Component
//@RocketMQMessageListener(consumerGroup = ConsumerGroupConstant.PRODUCT_SUBSCRIBE_GROUP, topic = BaseServerTopicConstants.PRODUCT_SUBSCRIBE_TOPIC, selectorExpression = BaseServerTopicConstants.DEVICE_STATE_CHANGE_TAG + "||" + BaseServerTopicConstants.DEVICE_LIFE_CYCLE_CHANGE_TAG + "||" + BaseServerTopicConstants.DEVICE_REPORT_INFO_TAG, messageModel = MessageModel.BROADCASTING)
//public class ProductSubscribeMessageRocketmqConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {
//
//    @Autowired
//    private RedisCached redisCached;
//
//    @Async("travelAsync")
//    @Override
//    public void onMessage(MessageExt ext) {
//        if (ext == null) {
//            log.info("产品订阅消息为空topic：{}", BaseServerTopicConstants.PRODUCT_SUBSCRIBE_TOPIC);
//            return;
//        }
//        String message = new String(ext.getBody());
//        log.info("产品订阅消息：{}", message);
//        Rp rp = JSON.parseObject(message, Rp.class);
//        String tag = rp.getTag();
//        Object data = rp.getData();
//        Long productId = rp.getProductId();
//        DeviceDO deviceDO = null;
//        DeviceAttributeVO deviceAttributeVO = null;
//        if (productId == null) {
//            log.info("产品订阅————产品ID为空");
//            return;
//        }
//        try {
//            if (data instanceof DeviceDO) {
//                deviceDO = (DeviceDO) data;
//                if (deviceDO == null) {
//                    log.info("设备消息体为空: {}", rp);
//                    return;
//                }
//            } else if (data instanceof DeviceAttributeVO) {
//                deviceAttributeVO = (DeviceAttributeVO) data;
//                if (deviceAttributeVO == null) {
//                    log.info("设备属性消息体为空: {}", rp);
//                    return;
//                }
//            } else {
//                log.info("产品订阅消息错误：{}", data);
//                return;
//            }
//            String str = redisCached.get(RedisConstant.PRODUCT_SUBSCRIBE_INFO_KEY + productId);
//            if (StringUtils.isNotBlank(str)) {
//                SubscribeDO subscribeDO = JSON.parseObject(str, SubscribeDO.class);
//                if (subscribeDO == null) {
//                    log.info("redis不为空但产品订阅为空：{}", str);
//                    return;
//                }
//                List<Integer> pushTypes = Arrays.stream(subscribeDO.getPushType().split(",")).map(Integer::parseInt).collect(Collectors.toList());
//                HttpMessagePushVO messagePushVO = new HttpMessagePushVO();
//                switch (tag) {
//                    //设备状态变更tag
//                    case BaseServerTopicConstants.DEVICE_STATE_CHANGE_TAG:
//                        if (pushTypes.contains(NumericalConstants.ONE)) {
//                            messagePushVO.setMessageType(NumericalConstants.ONE);
//                            messagePushVO.setDeviceAttributeVO(deviceAttributeVO);
//                        }
//                        break;
//                    //设备生命周期变更tag
//                    case BaseServerTopicConstants.DEVICE_LIFE_CYCLE_CHANGE_TAG:
//                        if (pushTypes.contains(NumericalConstants.TWO)) {
//                            messagePushVO.setMessageType(NumericalConstants.TWO);
//                            messagePushVO.setDeviceDO(deviceDO);
//                        }
//                        break;
//                    //设备上报信息tag
//                    case BaseServerTopicConstants.DEVICE_REPORT_INFO_TAG:
//                        if (pushTypes.contains(NumericalConstants.THREE)) {
//                            messagePushVO.setMessageType(NumericalConstants.THREE);
//                            messagePushVO.setDeviceAttributeVO(deviceAttributeVO);
//                        }
//                        break;
//                    default:
//                        log.info("未知产品订阅tag: {}", tag);
//                        return;
//                }
//                String url = subscribeDO.getUrl();
//                HttpUtil.post(url, Map.of("message", JSON.toJSONString(messagePushVO)));
//            }
//        } catch (Exception e) {
//            log.error("产品订阅数据-->消费失败，失败原因: {}", ExceptionUtil.getStackStr(e));
//        }
//    }
//
//    @Override
//    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
//        // 设置消费者消息重试次数
//        defaultMQPushConsumer.setMaxReconsumeTimes(3);
//        //设置实例名称
//        defaultMQPushConsumer.setInstanceName(MqttTopicPowerConstants.PRODUCT_SUBSCRIBE_INSTANCE_NAME);
//    }
//}